package com.zf.completable_future;

import org.junit.jupiter.api.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * {@link java.util.concurrent.CompletableFuture}
 * CompletableFuture 提供了简单易用且完善的异步API，我们可以使用这些API轻松完成复杂的异步操作编写，
 * 异步任务的执行默认使用CompletableFuture内部的 ForkJoinPool线程池对象来执行。
 */
public class CompletableFutureTest {

    /**
     * runAsync中提交的任务是没有返回值的，传参是Runnable接口，无返回值
     */
    @Test
    void testRunAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " 执行了");
        });
        // 阻塞获取执行结果
        Void unused = completableFuture.get(); // null
        System.out.println(unused);
    }

    /**
     * supplyAsync中提交的任务是必须要有返回值的，传参是Supplier接口，有返回值
     */
    @Test
    void testSupplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return Thread.currentThread().getName();
        });

        // 阻塞获取执行结果
        String res = future.get();
        System.out.println(res);
    }


    /**
     * 如果提交的任务还没有执行完成，则complete()方法会立刻结束任务的执行并返回true，get()方法会返回complete()方法传入的参数。
     * 如果提交的任务已经完成了，则complete()方法返回false，get()方法会返回任务的执行结果。
     */
    @Test
    void testComplete() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return Thread.currentThread().getName();
        });

        Thread.sleep(4000);
        boolean status = future.complete("直接结束");
        System.out.println(status);

        String res = future.get();
        System.out.println(res);
    }


    /**
     * 如果提交的任务还没有执行完成，则completeExceptionally()方法会立刻结束任务的执行并返回true，get()方法会抛出completeExceptionally()传入的异常。
     * 如果提交的任务已经完成了，则completeExceptionally()方法返回false，get()方法会返回任务的执行结果。
     */
    @Test
    void testCompleteExceptionally() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return Thread.currentThread().getName();
        });

        Thread.sleep(4000);
        boolean status = future.completeExceptionally(new RuntimeException("出错了"));
        System.out.println(status);

        String res = future.get(); // 直接结束
        System.out.println(res);
    }


    /**
     * whenComplete 当上一个任务执行完成后会触发 紧随其后的whenComplete中的任务的执行，多个 whenComplete 依然可以获取 whenComplete 之前的任务的结果和异常。
     * whenComplete：继续使用上一个任务的线程
     * whenCompleteAsync：从线程池中去一个线程（也可能取到和上一个任务使用的线程）来执行whenCompleteAsync中的任务，可以使用默认线程池，也可以手动指定线程池
     * <p>
     * 其他的 有Async 和 没有Async 的方法也是同样的意思
     */
    @Test
    void testWhenComplete() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // // 如果任务内部产生了异常，那么该任务里后续代码不再执行，但是后续的所有的 whenComplete、exceptionally、handle 中的任务依然可以执行，
            // 如果异常没有被处理，则会在通过get()方法获取最终结果时会抛出该异常
            // int a = 1 / 0;
            System.out.println("1-thread: " + Thread.currentThread().getName());
            return "结果0";
        }).whenComplete((res, err) -> {
            System.out.println("2-res: " + res + ", thread: " + Thread.currentThread().getName());
        }).whenCompleteAsync((res, err) -> {
            System.out.println("3-res: " + res + ", thread: " + Thread.currentThread().getName());
        }).whenCompleteAsync((res, err) -> {
            System.out.println("4-res: " + res + ", thread: " + Thread.currentThread().getName());
        }, Executors.newFixedThreadPool(2));

        String res = future.get();
        System.out.println("外部获取：" + res);
    }


    /**
     * exceptionally 方法接收前面所有任务中第一个产生的异常，然后返回一个正常值，get()方法获取 exceptionally方法中任务返回的值。
     * exceptionally中也能抛出异常。
     * 如果前面的任务没有产生异常，或者异常已经被处理了，则不会在进入 exceptionally 中。
     * <p>
     * 只有 exceptionally 才能处理异常，其他方法依然会把异常传递出去。
     */
    @Test
    void testExceptionally() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("1-thread: " + Thread.currentThread().getName());
            int a = 1 / 0;
            return "结果0";
        }).whenComplete((res, err) -> {
            System.out.println("2: ");
            // throw new RuntimeException("2自定义异常了");
        }).whenComplete((res, err) -> {
            System.out.println("3: ");
            // throw new RuntimeException("3自定义异常了");
        }).exceptionally(err -> {
            System.out.printf("4: 异常是：%s\n", err.getCause());
            return "错误了1";
        }).exceptionally(err -> {
            return "错误了2";
        });

        String res = future.get();
        System.out.println("外部获取：" + res);
    }


    /**
     * handle：无论前面的步骤是否发生异常都会被调用，可以在内部处理掉异常并返回正常值，也可以抛出异常
     */
    @Test
    void testHandle() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("1-thread: " + Thread.currentThread().getName());
            int a = 1 / 0;
            return "结果0";
        }).handle((res, err) -> {
            System.out.println("h1:");
            return "修复了错误";
        }).whenComplete((res, err) -> {
            System.out.println("2: ");
        }).exceptionally(err -> {
            System.out.printf("3: 异常是：%s\n", err.getCause());
            return "错误了1";
        });

        String res = future.get();
        System.out.println("外部获取：" + res);
    }


    /**
     * thenApply 可以对上一步的结果进行再处理，返回一个结果给下一步使用，
     * 一旦 提交的任务 或者 thenApply 中的任务发生了异常则不会再执行之后的任务，如果后续有handle、exceptionally、whenComplete方法则会执行这些方法中的任务。
     */
    @Test
    void testThenApply() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("1-thread: " + Thread.currentThread().getName());
            return "结果0";
        }).thenApply(res -> {
            System.out.println("1: " + res);
            int a = 1 / 0;
            return res + "1";
        }).thenApply(res -> {
            System.out.println("2: " + res);
            return res + "2";
        }).exceptionally(err -> {
            System.out.println("错误了1");
            return "异常1";
        });

        String res = future.get();
        System.out.println("外部：" + res);
    }

    /**
     * thenAccept 可以对上一步的结果进行再处理，但是不再返回任何值
     * 一旦 提交的任务 或者 thenAccept 中的任务发生了异常则不会再执行之后的任务，如果后续有handle、exceptionally、whenComplete方法则会执行这些方法中的任务。
     */
    @Test
    void testThenAccept() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("1-thread: " + Thread.currentThread().getName());
            return "结果0";
        }).thenAccept(res -> {
            System.out.println("1: " + res);
            int a = 1 / 0;
        }).thenAccept(res -> {
            System.out.println("2: " + res); // null
        }).exceptionally(err -> {
            System.out.println("错误了1");
            return null;
        });
    }

    /**
     * thenRun 标识上一步执行结束后就执行该方法中的任务，不依赖上一步的结果，也没有返回值
     * 一旦 提交的任务 或者 thenRun 中的任务发生了异常则不会再执行之后的任务，如果后续有handle、exceptionally、whenComplete方法则会执行这些方法中的任务。
     */
    @Test
    void testThenRun() {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("1-thread: " + Thread.currentThread().getName());
            return "结果0";
        }).thenRun(() -> {
            System.out.println("1");
            int a = 1 / 0;
        }).thenRun(() -> {
            System.out.println("2");
        }).exceptionally(err -> {
            System.out.println("错误了1");
            return null;
        });
    }

    /**
     * thenCompose 如果是方法内部返回了 CompletableFuture ,
     * 那么thenCompose会去除最外层的CompletableFuture,将内部返回的结果封装成新的CompletableFuture，然后执行下一步
     */
    @Test
    void testThenCompose() throws ExecutionException, InterruptedException {
        CompletableFuture<CompletableFuture<String>> plain = CompletableFuture.supplyAsync(() -> {
            return "1-plain";
        }).thenApply(res -> {
            // 返回一个 CompletableFuture
            return CompletableFuture.supplyAsync(() -> {
                return "2-plain";
            });
        });

        // 需要get两次
        String res1 = plain.get().get();
        System.out.println(res1);


        CompletableFuture<String> compose = CompletableFuture.supplyAsync(() -> {
            return "1-compose";
        }).thenCompose(res -> {
            // 返回一个 CompletableFuture
            return CompletableFuture.supplyAsync(() -> {
                return "2-compose";
            });
        });
        // 只需要get一次
        String res2 = compose.get();
        System.out.println(res2);
    }


    /**
     * thenCombine 获取上一步的结果和本次任务的结果，然后交给处理
     */
    @Test
    void testThenCombine() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "1";
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            return "2";
        }), (res1, res2) -> {
            return res1 + res2;
        });

        String res = future.get();
        System.out.println(res);
    }

    /**
     * allOf会返回一个新的 CompletableFuture, 当所有的CompletableFuture都完成时，allOf返回的CompletableFuture也完成了。
     * 只检查supplyAsync线程是否完成了，如果后续又开启了新的线程比如 thenApplyAsync , 则不会检查新的线程
     */
    @Test
    void testAllOf() {
        CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> {
            System.out.println("1");
            return "1";
        });
        CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> {
            System.out.println("2");
            return "2";
        });
        CompletableFuture<String> c = CompletableFuture.supplyAsync(() -> {
            System.out.println("3");
            return "3";
        });

        CompletableFuture<Void> allOf = CompletableFuture.allOf(a, b, c);
        // 等待所有的CompletableFuture完成
        allOf.join();
        System.out.println("都完成了");
    }


    /**
     * allOf会返回一个新的 CompletableFuture, 当任意一个CompletableFuture都完成时，anyOf就那个完成了的CompletableFuture的结果。
     * 只检查supplyAsync线程是否完成了，如果后续又开启了新的线程比如 thenApplyAsync , 则不会检查新的线程
     */
    @Test
    void testAnyOf() throws ExecutionException, InterruptedException {
        CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> {
            System.out.println("1");
            try {
                TimeUnit.SECONDS.sleep(4L);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "1";
        });
        CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> {
            System.out.println("2");
            try {
                TimeUnit.SECONDS.sleep(5L);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            return "2";
        });
        CompletableFuture<String> c = CompletableFuture.supplyAsync(() -> {
            System.out.println("3");
            try {
                TimeUnit.SECONDS.sleep(3L);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "3";
        });

        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(a, b, c);
        // 等待任意一个CompletableFuture执行完成
        anyOf.join();
        // 返回任意一个CompletableFuture完成后的结果
        Object res = anyOf.get();
        System.out.println("都完成了: " + res);
    }


}
